package com.study.skywalking.component;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * Created by zhangbo on 2020/6/10.
 */
@Slf4j
@Component
public class MqConsumer implements InitializingBean {

    @Value("${ROCKET.CONSUMER.GROUP:studySkywalking}")
    private String consumerGroup;

    @Value("${ROCKET.NAME.SERVER}")
    private String nameServer;

    @Value("${ROCKET.TOPIC:MyTopic}")
    private String topic;

    private DefaultMQPushConsumer consumer;


    public void init() {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameServer);
        try {
            consumer.subscribe(topic, "*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    MessageExt messageExt = list.get(0);
                    try {
                        String messageInfo = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        log.info("消费端接口收到消息key:{},value:{}", messageExt.getKeys(), messageInfo);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } catch (Exception e) {
                        log.error("消息消费异常：{}", e);
                        if (messageExt.getReconsumeTimes() >= 3) {
                            //已经重试3次了，不再做处理了，在此处记录日志做补偿机制，返回成功
                            log.error("message:" + messageExt.getKeys());
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
            });
            consumer.start();
        } catch (MQClientException e) {
            log.error("MQ消费端异常：{}", e);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        init();
    }
}
